package org.jeecg.modules.iot.middleware.kafka.consumer;

import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.jeecg.common.base.BaseMap;
import org.jeecg.common.modules.redis.client.JeecgRedisClient;
import org.jeecg.common.util.RedisUtil;
import org.jeecg.jt808.JTT808Decoder;
import org.jeecg.jt808.req.TransparencyMsg;
import org.jeecg.modules.iot.cache.EBikeCache;
import org.jeecg.modules.iot.middleware.processor.control.TaitModbusControlParser;
import org.jeecg.modules.iot.middleware.processor.powerOperate.TaitModbusParser;
import org.jeecg.modules.iot.middleware.service.IConsumptionService;
import org.jeecg.modules.iot.middleware.service.ILocationService;
import org.jeecg.modules.iot.middleware.service.IPowerService;
import org.jeecg.modules.iot.middleware.util.AddressUtil;
import org.jeecg.modules.iot.middleware.util.JsonUtil;
import org.jeecg.modules.iot.model.Consumption;
import org.jeecg.modules.iot.model.Location;
import org.jeecg.modules.iot.model.MqttMessageJson;
import org.jeecg.modules.iot.model.control.TaitModbusControllerResponse;
import org.jeecg.modules.iot.model.power.TaiPower;
import org.jeecg.modules.iot.model.power.TaitModbusPowerResponse;
import org.jeecg.modules.websocket.constant.WsMessageConstant;
import org.jeecg.modules.websocket.enums.WsBussinesType;
import org.mybatis.logging.Logger;
import org.mybatis.logging.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.stereotype.Component;
import org.springframework.util.backoff.FixedBackOff;

import java.io.IOException;
import java.sql.Timestamp;
import java.time.Duration;
import java.time.Instant;
import java.util.Date;
import java.util.concurrent.CompletableFuture;

import static org.jeecg.modules.iot.cache.EBikeCache.roundToTwoDecimalPlaces;

@RequiredArgsConstructor
@Component
public class TransparencyConsumer {

    private static final Logger log = LoggerFactory.getLogger(TransparencyConsumer.class);
    ILocationService locationService;
    @Autowired
    IPowerService powerService;
    private final ObjectMapper objectMapper = new ObjectMapper();
    private Instant lastBMSEntryTime = null;
    @Autowired
    TaitModbusParser taitModbusParser;
    @Autowired
    TaitModbusControlParser taitModbusControlParser;

    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private AddressUtil addressUtil;

    @Autowired
    private IConsumptionService consumptionService;


    private static final String NATIONAL_STANDARD_SPEED_LIMIT_KEY = "vehicle:%s:nationalStandardSpeedLimit"; //  Redis key 格式


    @Autowired
    EBikeCache eBikeCache;
    @Autowired
    private JeecgRedisClient jeecgRedisClient;

    @KafkaListener(topics = "${jeecg.kafka.transparent.topic}", groupId = "${spring.kafka.consumer.group-id}", concurrency = "12")
    public void listen(ConsumerRecord<String, String> record) throws IOException {
        // 处理透传消息
        if (!"865376074447929".equals(record.key())){
            return;
        }
        MqttMessageJson mqttMessageJson = objectMapper.readValue(record.value(), MqttMessageJson.class);

        TransparencyMsg jt808DataPacket = (TransparencyMsg) JTT808Decoder.decode(mqttMessageJson.getData());

        // ... 处理透传消息的逻辑 ...

        // 解析和处理透传消息
        processTransparencyMsg(jt808DataPacket, record.key(), mqttMessageJson.getSeq());




    }

    @Bean
    public SeekToCurrentErrorHandler TransparencyError(KafkaOperations<Object, Object> template) {
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
        return new SeekToCurrentErrorHandler(recoverer, new FixedBackOff(0L, 0L)); // Retry 3 times with 1-second intervals
        // ... 错误处理，失败一次的数据直接发送到死信 ...
    }


    private void processTransparencyMsg(TransparencyMsg jt808DataPacket, String deviceId, String seq) throws IOException {
        ByteBuf buf = Unpooled.wrappedBuffer(jt808DataPacket.getTransparentMsgContent());


        int slaveAddressTrans = buf.readByte() & 0xFF;
        buf.release();

        // 异步处理不同类型的透传消息
        CompletableFuture.runAsync(() -> {
            try {
                if (slaveAddressTrans == 0x03) { // BMS
                    handleBmsData(jt808DataPacket, deviceId, seq);
                } else if (slaveAddressTrans == 0x04) { // 控制器
                    handleControllerData(jt808DataPacket, deviceId, seq);
                } else {
                    // 处理其他从机地址或错误情况
                    System.err.println("未知的从机地址: " + slaveAddressTrans);
                }
            } catch (Exception e) {

                // ... 错误处理 ...
            }
        });

    }


    private void handleBmsData(TransparencyMsg jt808DataPacket, String deviceId, String seq) throws IOException {
        Integer startAdress = addressUtil.getAddressAndDelete(deviceId, seq);
        TaitModbusPowerResponse tai = taitModbusParser.parseBatteryData(jt808DataPacket.getTransparentMsgContent(), startAdress);
        Instant currentBMSEntryTime = Instant.now();
        if (lastBMSEntryTime != null) {
            Duration duration = Duration.between(lastBMSEntryTime, currentBMSEntryTime);

        }
        lastBMSEntryTime = currentBMSEntryTime;

        if (tai != null) {
            if (tai.getSlaveAddress() == 0x03 && tai.getFunctionCode() == 0x03) {
                TaiPower taiPower = new TaiPower();
                TaiPower power = powerService.queryLastByDeviceId(deviceId);
                taiPower.setVoltage(roundToTwoDecimalPlaces(tai.getBatteryPackVoltage()));
                taiPower.setCurrent(roundToTwoDecimalPlaces(tai.getBatteryCurrent()));
                if (power == null) {
                    power = new TaiPower();
                }

                Location location = locationService.queryByDeviceId(jt808DataPacket.getHeader().getTerminalPhone());

                Float rP = (float) (tai.getBatteryPackVoltage() * 0.95);
                //System.out.println("高位："+highCount +" 低位："+lowCount+"电池电压"+tai.getBatteryPackVoltage());
                taiPower.setRPow(roundToTwoDecimalPlaces(rP * tai.getBatteryRemainingCapacity() / 1000));
                taiPower.setSoh(tai.getBatteryPackHealth());
                //System.out.println(tai.getBatteryRemainingCapacity()+"--------------剩余容量----------"+taiPower.getRPow());
                taiPower.setTs(new Timestamp(new Date().getTime()));
                /*剩余电量*/
                taiPower.setRPowCap(roundToTwoDecimalPlaces((float) tai.getBatteryRelativeCapacity()));
                taiPower.setDeviceId(deviceId);

                if (taiPower.getRPow() == 0f) {
                    return;
                }


                if (power == null) {
                    power = new TaiPower();

                }

                eBikeCache.setVehiclePower(deviceId,taiPower);
                updateEPower(taiPower, power, location, tai);
                eBikeCache.updateVehicleBatteryLevel(deviceId, Double.valueOf(taiPower.getRPowCap()));

                String powerMessage = JsonUtil.createPubJson(WsBussinesType.POWER.getCode(), true, taiPower);
                powerService.add(taiPower);
                //sseEmitterManager.sendMessageToEmitter(mqttMessageJson.getId(), BussinesType.ALLINFO.getType(), powerMessage);

                BaseMap poweParams = new BaseMap();
                poweParams.add("terminalCode", deviceId);
                poweParams.add("message", powerMessage);
                jeecgRedisClient.sendMessage(WsMessageConstant.EBIKE_REALTIME_MESSAGE, poweParams);
            }
        }
    }

    private void handleControllerData(TransparencyMsg jt808DataPacket, String deviceId, String seq) {
        Integer startAdress = addressUtil.getAddressAndDelete(deviceId, seq);
        TaitModbusControllerResponse taitModbusControllerResponse = taitModbusControlParser.parseControllerData(jt808DataPacket.getTransparentMsgContent(), startAdress);

        if (taitModbusControllerResponse != null) {
            if (taitModbusControllerResponse.isNationalStandardSpeedLimit()) {
                String redisKey = String.format(NATIONAL_STANDARD_SPEED_LIMIT_KEY, deviceId);
                redisUtil.set(redisKey, taitModbusControllerResponse.isNationalStandardSpeedLimit());
            }
        }
    }

    private void updateEPower(TaiPower taiPower, TaiPower power, Location location, TaitModbusPowerResponse tai) {
        //FieldStatus fieldStatus = new FieldStatus(location.getStatusField());

        String deviceId = taiPower.getDeviceId();
        Float x = power.getRPow() - taiPower.getRPow();

        // 如果 RPow 为 0，则将 EPower 设为 0
        if (power.getRPow() == 0) {
            taiPower.setEPower(0f);
        } else {
            // 如果电流为负且功率差为正，更新 EPower
            if (taiPower.getCurrent() < 0 && x > 0) {
                taiPower.setEPower(
                        power.getEPower() == null ? 0f : power.getEPower()
                );
                taiPower.setEPower(taiPower.getEPower() + x);
            } else {
                // 否则直接继承 power 的 EPower
                taiPower.setEPower(power.getEPower());
            }
        }
        if (location == null) {
            return;
        }

        if (x > 0 && taiPower.getCurrent() < 0 && power.getCurrent() < 0 && location.getSnnNumber() > 3) {
            float fPower = (power.getFpower() == null ? 0f : power.getFpower()) + x;
            taiPower.setFpower(fPower);
        } else {
            if (taiPower.getFpower() != null && power.getFpower() != null) {
                taiPower.setFpower(power.getFpower());
            } else {
                taiPower.setFpower(0f);
            }

        }

        if (taiPower.getFpower() > 10 && taiPower.getCurrent() < 0 && power.getCurrent() < 0) {
            Consumption lastConsumption = consumptionService.queryLatestByDeviceId(deviceId);
            // 创建新的能耗记录
            Consumption consumption = new Consumption();
            consumption.setDeviceId(deviceId);
            consumption.setTs(location.getTs());  // 使用当前 Location 的时间戳
            consumption.setFpower(taiPower.getFpower());  // 设置当前行驶里程
            consumption.setEpower(x);
            if (lastConsumption != null) {// 设置新消耗的电量
                Location prev500Location = locationService.queryNearestLocationBeforeTime(deviceId, lastConsumption.getTs());
                float energy500Mil = location.getMileage() - prev500Location.getMileage();
                // 如果存在上一次的能耗记录，则计算能耗差值
                if (energy500Mil > 0) {


                    consumption.setFMileage(energy500Mil);
                    float energy500Consumption = consumption.getFpower() / energy500Mil; // 计算 500 米的能耗
                    if (energy500Consumption == 0) {
                        consumption.setEConsum(lastConsumption.getEConsum());
                    } else {
                        consumption.setEConsum(energy500Consumption);
                    }


                }
            } else {


             /*   RentalCarQueryVo rentalCarQueryVo = new RentalCarQueryVo();
                rentalCarQueryVo.setCarCode(deviceId);

                Result<List<RentalCar>> result = iRentalCarAPI.listByQuery(rentalCarQueryVo);
                if (result != null && result.isSuccess() && result.getResult() != null) {
                    List<RentalCar> existingCars = result.getResult();
                    if (existingCars.size() > 0) {
                        Long carTypeId = existingCars.get(0).getCarTypeId();
                        if (carTypeId != null) {
                            RentalCarType carType = iRentalCarAPI.selectCarTypeById(carTypeId);
                            if (carType != null && carType.getInitialConsumption() != null) {
                                consumption.setEConsum(carType.getInitialConsumption());
                            } else {
                                consumption.setEConsum(0f);
                            }

                        }
                    }

                }*/
            }
            // 添加新的能耗记录
            if (consumption.getEConsum() > 0.1f) {

                consumption.setEConsum(lastConsumption.getEConsum());
            }
            consumptionService.add(consumption);

            // 重置 `FPow`，因为超过 50 之后我们已经记录了消耗
            taiPower.setFpower(0f);

        }
    }

}